package realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import realtime.util.MyKafkaUtil;

/**
 * @author MengX
 * @create 2023/3/25 08:54:55
 */
public class dwd_review_info {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String groupId = "dwd_review_info";
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL(groupId));
        Table reviewInfo = tableEnv.sqlQuery("select \n" +
                "    data['id'] id,\n" +
                "    data['user_id'] user_id,\n" +
                "    data['course_id'] course_id,\n" +
                "    data['review_txt'] review_txt,\n" +
                "    data['review_stars'] review_stars,\n" +
                "    data['create_time'] create_time,\n" +
                "    data['deleted'] deleted \n" +
                "from `topic_db`\n" +
                "where `table` = 'review_info'  and  `type` = 'insert'");

        tableEnv.createTemporaryView("review_info",reviewInfo);

        tableEnv.executeSql("" +
                "CREATE TABLE dwd_review_info (\n" +
                "  `id` STRING,\n" +
                "  `user_id` STRING,\n" +
                "  `course_id` STRING,\n" +
                "  `review_txt` STRING,\n" +
                "  `review_stars` STRING,\n" +
                "  `create_time` STRING,\n" +
                "  `deleted` STRING\n" +
                ") " + MyKafkaUtil.getKafkaSinkDDL("dwd_review_info"));

        tableEnv.executeSql("insert into dwd_review_info select * from review_info");

    }
}
